package com.sinux.generality.basesupport.utils.kafka.consumer;

import com.sinux.generality.basesupport.utils.kafka.producer.DefaultKafkaProducerClass;
import com.sinux.generality.basesupport.utils.kafka.producer.KafkaProducerInterface;
import com.sinux.generality.basesupport.utils.kafka.producer.SimpleKafkaProducerClass;

import java.util.List;
import java.util.Map;

public class ConsumerFactory {

    public static final String buildName_default = "default";
    public static final String buildName_simple = "simple";

    public static KafkaConsumerInterface buildConsumerClassByBoostrapServers(String buildName, List<String> boostrapServers, String groupID) {
        KafkaConsumerInterface result = null;
        if(buildName_default.equals(buildName)){
            return new DefaultKafkaConsumerClass(boostrapServers,groupID);
        }else if(buildName_simple.equals(buildName)){
            return new SimpleKafkaConsumerClass(boostrapServers,groupID);
        }
        return result;
    }

    public static KafkaConsumerInterface buildConsumerClassByConfigMap(String buildName, Map<String,Object> configMap) {
        KafkaConsumerInterface result = null;
        if(buildName_default.equals(buildName)){
            return new DefaultKafkaConsumerClass(configMap);
        }
        else if(buildName_simple.equals(buildName)){
            return new SimpleKafkaConsumerClass(configMap);
        }
        return result;
    }

    public static KafkaConsumerInterface buildConsumerClassByObject(String buildName,Object o) {
        KafkaConsumerInterface result = null;
        if(buildName_default.equals(buildName)){
            return new DefaultKafkaConsumerClass((DefaultKafkaConsumerClass) o);
        }
        else if(buildName_simple.equals(buildName)){
            return new SimpleKafkaConsumerClass((SimpleKafkaConsumerClass) o);
        }
        return result;
    }
}
